package cn.dmp.report

import cn.dmp.beans.Log
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}


/**
  * 需求三：
  * 利用spark算子的方式进行离线数据统计:还是统计各省市的数据量分布情况
  *
  */
object ProCityRpt_Version3 {

  def main(args: Array[String]): Unit = {

    // 0.判断参数
    if (args.length != 2){
      println(
        """
          |cn.dmp.report.ProCityRpt_Version3
          |参数：
          |   logInputPath
          |   resultOutputPath
        """.stripMargin)
    }

    // 1.接收程序参数
    val Array(logInputPath, resultOutputPath) = args

    // 2.创建SparkSession
    val sparkSession: SparkSession = SparkSession.builder().appName("ProCityRpt_Version3").master("local[*]")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //spark优化：spark.serializer修改磁盘序列化方式
      .enableHiveSupport()
      .getOrCreate()

    /**
      * import sparkSession.implicits._  !!!!!!!!!处理DataSet类型数据必须加的包！！！
    val res: RDD[((String, String), Int)] = sparkSession.read.textFile(logInputPath)
      .map(line => line.split(",", -1)).filter(_.length >= 85)
      .map(arr => ((arr(24), arr(25)), 1)) //val log = Log(arr); ((log.Provincename, log.Cityname), 1)
      .rdd //这里将dataset数据转换成rdd，是为了使用reduceByKey方法，因为dataset类型没有reduceByKey方法只有reduce方法
      .reduceByKey(_ + _)
      .map(t => t._1._1 + "," + t._1._2 + "," + t._2)

    /**
      * 3.用spark读取数据:读取什么数据：
      * 2016-10-01_06_p1_invalid.1475274123982.log.FINISH
       */
    // 保存结果数据到磁盘
    res.saveAsTextFile(resultOutputPath)
    */

    /**
      * //想不通这里运行为啥会报错？？？？？？？？？是因为没有处理DataSet类型数据必须加的包！！！import sparkSession.implicits._
      * Unable to find encoder for type stored in a Dataset.  Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._  Support for serializing other types will be added in future releases.
      * val res: Dataset[Array[String]] = sparkSession.read.textFile(logInputPath).map(line => line.split(","))
      * DataSet,DataFrame,RDD这三个数据类型分别是谁得到的？？？
      * SparkSession，SparkContext,这两个区别？？？分别是用来读取哪些类型的文件？？
      * SparkSession可以读取pqrquet类型的文件：read.parquet(logInputPath)，但是好像读取正常的a.log.FINISH日志文件后不能做map(_.split(","))操作。
      *
      * 搞懂了！！！！！！！！：报上面的错是因为要想使用DataSet类型数据，必须导入这个方法：import sparkSession.implicits._  注意这个sparkSession是自己上面定义上下文的变量名。
      * sparkSession.read.textFile(logInputPath)返回的是DataSet类型的数据。
      *
      * 搞不懂所以我就用下面SparkContext的textFile读取文件的方法 了。
      *
      */
    /**
      * 结论：
      * 好像sparkSession的read.textFile不能读取log类型文件进行map处理？？？
      * (是可以的：sparkSession的read.textFile返回的是DataSet类型的数据，该类型的数据处理必须要导入一个包import sparkSession.implicits._)
      * （sparkSession的read.parquet方法读取后返回的是DataFrame类型， DataFrame类型map函数里面的是Row类型数据，而不是String类型数据可以直接split(“,”)）
      * （Row对象把数据给包装起来了而已map(f：(Row)=>func))， 我们可以用row => {row.get(24),row.getAs[Int](“Requestmode”)。这两种方法都可以获取到该行对象的数据Int或者String。}。）
      *
       */
    val sc: SparkContext = sparkSession.sparkContext
    val res: RDD[String] = sc.textFile(logInputPath)
      .map(line => line.split(",", -1)).filter(_.length >= 85)
      .map(arr => ((arr(24), arr(25)), 1)) //val log = Log(arr); ((log.Provincename, log.Cityname), 1)
      .reduceByKey(_ + _)
      .map(t => t._1._1 + "," + t._1._2 + "," + t._2)
    //将((湖北省,荆州市),214)变成： 湖北省,荆州市,214 这种的结构化的数据
    res.saveAsTextFile(resultOutputPath)


    // 4.

    // ?.关闭
    sc.stop()
    sparkSession.stop()

    //本地测试运行：!!!!!注意提前设置好参数：输入目录和输出目录
    /**
      * 输入目录：E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\data\2016-10-01_06_p1_invalid.1475274123982.log.FINISH
      * 输出目录: E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\procityrpt_Version3
      *
      */

  }

}
